package spark.work

//import breeze.linalg.{max, sum}
//import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import com.kongming.utils.misc.StringUtil
import org.apache.commons.lang.StringUtils
import org.apache.spark.sql.types._
import org.apache.spark.sql.{Row, SQLContext, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}
import spark.work.Sampling.typeModifyUdf
import util.DataSampleUtils

import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
//import org.apache.spark.sql.functions._
import org.apache.spark.sql.functions.{col, udf}

/**
  * Created by liuwei on 2017/11/8.
  */
object ClearingRepeat {

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setAppName("ClearingRepeat").setMaster("local[4]")
    val sc = new SparkContext(sparkConf)
    val sqlContext = new SQLContext(sc)
    val spark = SparkSession.builder

    val ss = spark.getOrCreate();
    val df = ss.createDataFrame(Seq(
//      ("张三", "女", "数学", 60, 15, "张老师"),
      ("张三", "女", "数学", 88, 15, ""),
//      ("张三", "女", "数学", 88, 15, "liu老师"),
      ("张三", "女", "语文", 70, 12, "李老师"),
      ("张三", "女", "物理", 50, 22, "张老师"),
      ("王二", "男", "数学", 50, 16, "赵老师"),
      ("王二", "男", "语文", 60, 20, "孙老师"),
      ("王二", "男", "历史", 50, 22, "李老师")
    )).toDF("姓名", "性别", "科目", "成绩", "排名", "监考老师")
    val titleNames = List("姓名", "性别", "科目", "成绩", "排名", "监考老师")
    val columnNames = List("姓名")
    val columnTagIdx = columnNames.map(titleNames.indexOf(_))
    val conditionName = "监考老师"
    val conditionTagIdx = titleNames.indexOf(conditionName)
    val filterValue = "nullValue"
    val filterType = "random2"
    val columnAddConditionName: List[String] = columnNames.filter(!_.equals(conditionName)) ::: List(conditionName)
    println("====" + columnAddConditionName)
    val dropKey = "eleColValNull"
    val columnDistinceValSeq = columnNames.map(column => {
      val tagArray = df.select(df.col(column)).distinct().collect()
      (column, tagArray.map(row => if (row.isNullAt(0)) dropKey else row.get(0).toString).toSeq)
    })
    val rdd = df.rdd.mapPartitions(rowIter => {
      val retIter = new ArrayBuffer[(List[String], String)]
      while (rowIter.hasNext) {
        val row = rowIter.next
        val key = columnTagIdx.map(row.get(_).toString)
        //.mkString("-")
        val value = row.get(conditionTagIdx).toString

        retIter += Tuple2(key, value.replace("","nullValue"))
      }
      retIter.iterator
    }).reduceByKey { case (left, right) => {
      if (null != left && left.equals(filterValue)) left
      else if (right.equals(filterValue)) right
      else if ("random".equals(filterType)) {
        left
      } else {
        ""
      }
    }
    }.filter(_._2 != "").map(s => {
      val arr = mutable.ArrayBuffer.empty[String]
      s._1.foreach(arr += _)
      arr += s._2
      Row.fromSeq(arr)
    })

    val schema = df.schema.filter(p => columnAddConditionName.contains(p.name))
    val schema2 = StructType(schema)

    val transformDF = ss.createDataFrame(rdd, schema2)
    //    val groupByCondition = df
    //      .groupBy(columnNames.head, columnNames.tail: _*)
    //      .agg(
    //        Map(
    //          s"$conditionName" -> s"max"
    //        )
    //      )
    //      .drop(conditionName)
    //      .toDF(columnAddConditionName: _*)
    //    groupByCondition.show
    //   var reserve = groupByCondition(columnAddConditionName.head) <=> df(columnAddConditionName.head)
    //    if (columnAddConditionName.size > 1) {
    //      columnAddConditionName.tail.foreach(each => {
    //        reserve = reserve.and(groupByCondition(each) <=> df(each))
    //      })
    //    }
    //    val columns = columnNames.map(col(_)).toSeq

    var df2 = transformDF.join(df, columnNames)
    columnAddConditionName.foreach(each => {
      df2 = df2.drop(df(each))
    })
    df2 = df2.dropDuplicates(columnAddConditionName.head, columnAddConditionName.tail: _*)
    df2.show()
    //    println(groupByCondition)
  }


  def convert(s: String, nRows: Int): String = {
    val len = s.length
    if (len == 0 || nRows < 2) return s
    var ret = ""
    val lag = 2 * nRows - 2
    //循环周期
    var i = 0
    while (i < nRows) {
      var j = i
      while (j < len) {
        ret += s.charAt(j)
        //非首行和末行时还要加一个
        if (i > 0 && i < nRows - 1) {
          val t = j + lag - 2 * i
          if (t < len) ret += s.charAt(t)
        }

        j += lag
      }
      {
        i += 1;
        i - 1
      }
    }
    ret
  }




}
